package com.example.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.example.common.redis.CacheService;
import com.example.common.util.R;
import com.example.model.schedule.dto.TaskDto;
import com.example.model.schedule.pojo.Taskinfo;
import com.example.model.schedule.pojo.TaskinfoLogs;
import com.example.schedule.mapper.TaskinfoLogsMapper;
import com.example.schedule.mapper.TaskinfoMapper;
import com.example.schedule.service.TaskInfoLogsService;
import com.example.schedule.service.TaskInfoService;
import io.swagger.v3.oas.annotations.Operation;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;

import java.util.Calendar;
import java.util.List;
import java.util.Set;

@Service
@Slf4j
public class TaskInfoServiceImpl extends ServiceImpl< TaskinfoMapper, Taskinfo > implements TaskInfoService {
    private static final String KEY_PREFIX = "FUTURE";
    @Autowired
    private TaskinfoMapper taskinfoMapper;
    @Autowired
    private TaskinfoLogsMapper taskinfoLogsMapper;
    @Autowired
    private CacheService cacheService;
    @Autowired
    private KafkaTemplate< String, String > kafkaTemplate;

    public boolean addTask(TaskDto taskDto) {
        Taskinfo taskinfo = new Taskinfo();
        BeanUtils.copyProperties(taskDto, taskinfo);
        taskinfoMapper.insert(taskinfo);

        // 设置taskId
        taskDto.setTaskId(taskinfo.getTaskId());

        TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
        BeanUtils.copyProperties(taskinfo, taskinfoLogs);
        taskinfoLogs.setStatus(TaskinfoLogs.SCHEDULED);
        taskinfoLogs.setVersion(1);
        taskinfoLogsMapper.insert(taskinfoLogs);

        // 判断执行时间，如果<=当前时间，立即执行（MQ），如果<=预设时间，写zset
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE, 5);
        if (taskinfo.getExecuteTime().getTime() <= System.currentTimeMillis()) {
            sendMessage(taskDto);
        } else if (taskinfo.getExecuteTime().getTime() <= calendar.getTime().getTime()) {
            String key = buildKey(taskDto);
            cacheService.zAdd(key, JSON.toJSONString(taskDto), taskinfo.getExecuteTime().getTime());
        }
        return true;
    }

    private void sendMessage(TaskDto taskDto) {
        // 发送消息，通知任务时间到
        log.info("发送需要执行的任务消息");
        kafkaTemplate.send("TASK_EXEC", JSON.toJSONString(taskDto));
        // 更新数据库，删除zset里面执行的数据
        taskinfoMapper.deleteById(taskDto.getTaskId());
        TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(taskDto.getTaskId());
        taskinfoLogs.setStatus(TaskinfoLogs.EXECUTED);
        taskinfoLogsMapper.updateById(taskinfoLogs);
        cacheService.zRemove(buildKey(taskDto), JSON.toJSONString(taskDto));

    }

    private String buildKey(TaskDto taskDto) {
        return KEY_PREFIX + taskDto.getTaskType() + ":" + taskDto.getPriority();
    }

    @Scheduled(cron = "0/20 * * * * ?")
    public void refreshZset() {
        log.info("开始执行从数据库读取最近5分钟中数据到zset");

        Set<String> keys = cacheService.scan(KEY_PREFIX + "*");
        cacheService.delete(keys);

        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE, 5);
        LambdaQueryWrapper<Taskinfo> wrapper = new LambdaQueryWrapper<>();
        wrapper.le(Taskinfo::getExecuteTime, calendar.getTime());
        List<Taskinfo> taskinfoList = taskinfoMapper.selectList(wrapper);
        if (CollectionUtils.isEmpty(taskinfoList)) {
            log.info("没有需要同步到zset的数据");
            return;
        }
        for (Taskinfo taskinfo : taskinfoList) {
            TaskDto taskDto = new TaskDto();
            BeanUtils.copyProperties(taskinfo, taskDto);
            String key = buildKey(taskDto);
            cacheService.zAdd(key, JSON.toJSONString(taskDto), taskinfo.getExecuteTime().getTime());
        }
    }

    @Scheduled(cron = "0/10 * * * * ?")
    public void checkZset() {
        log.info("开始执行检查zset任务是否需要执行");
        String token = cacheService.tryLock("checkZset", 20);
        if (token == null) {
            log.info("未获取到锁，不执行");
            return;
        }
        Set< String > futreKeys = cacheService.scan(KEY_PREFIX + "*");
        if (CollectionUtils.isEmpty(futreKeys)) {
            log.info("没有任务队列");
            return;
        }
        for (String futreKey : futreKeys) {
            Set< String > futreValues = cacheService.zRangeByScore(futreKey, 0, System.currentTimeMillis());
            if (CollectionUtils.isEmpty(futreValues)) {
                log.info("当前key: {}没有需要执行的任务消息", futreKey);
                continue;
            }
            log.info("当前需要执行的任务消息: {}", futreValues);
            for (String taskStr : futreValues) {
                sendMessage(JSON.parseObject(taskStr, TaskDto.class));
            }
        }
        log.info("检查zset任务执行结束");
    }
}
